package com.sunzm.dynamic_rule.engine;

import com.sunzm.dynamic_rule.pojo.LogBean;
import com.sunzm.dynamic_rule.pojo.ResultBean;
import com.alibaba.fastjson.JSON;
import org.apache.commons.collections.CollectionUtils;
import org.apache.commons.lang3.StringUtils;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.api.common.state.ListState;
import org.apache.flink.api.common.state.ListStateDescriptor;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.util.Collector;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.ArrayList;
import java.util.Map;
import java.util.Properties;
import java.util.regex.Pattern;

/**
 * 实时运营系统Demo
 * <p>
 * 规则：
 * 触发条件：addCart 事件（eventId=‘addCart’）
 * 画像属性条件： 性别为女(sex=1)  年龄小于30（age<30） 累计消费大于500(cst > 500>)
 * 行为属性条件: 加购物车（eventId=addCart, productId=8)>=2次, 点击广告（eventId=adClick,adId=6）>=1次
 * 行为次序条件：依次做过 加购物车（eventId=addCart, productId=8) -> 搜索事件(eventId=search)
 */
public class RuleEngineDemo {
    private static boolean isLocal = true;
    private static final Logger logger = LoggerFactory.getLogger(RuleEngineDemo.class);


    public static void main(String[] args) throws Exception {
        //获取执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        if (isLocal) {
            env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(new Configuration());
        }

        //kafka数据源
        Properties properties = new Properties();
        properties.setProperty(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "82.156.210.70:9093");
        properties.setProperty(ConsumerConfig.GROUP_ID_CONFIG, "flink-rule");
        //从最新的偏移量开始消费
        properties.setProperty(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "latest");
        //从最早的偏移量开始消费
        //properties.setProperty(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");

        //{"account":"0075","deviceId":"0075","eventId":"addCart","properties":{"productId":"8"},"timeStamp":1625327883237}
        //{"account":"0075","deviceId":"0075","eventId":"adClick","properties":{"adId":"6"},"timeStamp":1625327883238}
        //{"account":"0075","deviceId":"0075","eventId":"search","properties":{"searchWord":"裤子"},"timeStamp":1625327883239}
        DataStream<String> dataStream = env.addSource(new FlinkKafkaConsumer<>("applog", new SimpleStringSchema(), properties));

        //dataStream.print();
        //解析JSON格式的数据,把字符串格式一个变成LogBean对象
        //使用 flatMap 方法，可以在转换的同时完成过滤
        DataStream<LogBean> logBeanDataStream = dataStream.flatMap((String value, Collector<LogBean> out) -> {
            if (StringUtils.isNotBlank(value)) {
                //如果输入的数据不是空字符串才进行处理
                try {
                    LogBean logBean = JSON.parseObject(value, LogBean.class);
                    out.collect(logBean);
                } catch (Exception e) {
                    logger.error("JSON数据解析成LogBean对象异常:" + value, e);
                }
            }
        })
                //使用Java Lambda 表达式的写法， 无法自动推断类型，使用Scala就不存在这个问题，这一行就不需要写
                .returns(TypeInformation.of(LogBean.class));
        //.returns(LogBean.class);
        //.returns(Types.POJO(LogBean.class));

        //把数据按照设备Id进行keyBy（我们认为设备Id相同的是同一个用户）
        KeyedStream<LogBean, String> keyedStream = logBeanDataStream.keyBy(logBean -> logBean.getDeviceId());

        //开始进行规则判断
        SingleOutputStreamOperator<ResultBean> resultDataStream = keyedStream.process(
                //第一个参数类型为Key的类型, 第二个参数类型为输入数据的类型, 第三个参数的类型为 输出数据的类型
                new KeyedProcessFunction<String, LogBean, ResultBean>() {
                    private transient Connection conn = null;
                    private transient Table table = null;
                    //存放用户行为数据的状态
                    private ListState<LogBean> logBeanState = null;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        //open方法只会调用一次，避免多次创建HBase连接
                        org.apache.hadoop.conf.Configuration conf = HBaseConfiguration.create();
                        conf.set("hbase.zookeeper.quorum", "192.168.1.158:2181");
                        conn = ConnectionFactory.createConnection(conf);
                        table = conn.getTable(TableName.valueOf("user-profile"));

                        //定义一个List类型的State，存放用户行为数据
                        logBeanState = getRuntimeContext().getListState(new ListStateDescriptor<LogBean>(
                                "logBeanState", TypeInformation.of(LogBean.class)
                        ));
                    }

                    @Override
                    public void processElement(LogBean logBean, Context ctx, Collector<ResultBean> out) throws Exception {
                        //存入状态中
                        logBeanState.add(logBean);

                        //判断当前用户行为是否满足规则中的触发条件
                        //触发条件：addCart 事件（eventId=‘addCart’）
                        if (StringUtils.equals("addCart", logBean.getEventId())) {

                            //* 判断画像属性条件： 性别为女(sex=1)  年龄小于30（age<30） 累计消费大于500(cst > 500>)
                            //先从hbase中查询用户画像属性
                            //构建Get对象
                            Get get = new Get(Bytes.toBytes(logBean.getDeviceId()));
                            //设置需要取的列，如果不设置，就会全返回
                            get.addColumn(Bytes.toBytes("f"), Bytes.toBytes("sex"));
                            get.addColumn(Bytes.toBytes("f"), Bytes.toBytes("age"));
                            get.addColumn(Bytes.toBytes("f"), Bytes.toBytes("cst"));
                            //发起get查询
                            Result result = table.get(get);
                            //解析结果
                            String sex = Bytes.toString(result.getValue(Bytes.toBytes("f"), Bytes.toBytes("sex")));
                            String age = Bytes.toString(result.getValue(Bytes.toBytes("f"), Bytes.toBytes("age")));
                            String cst = Bytes.toString(result.getValue(Bytes.toBytes("f"), Bytes.toBytes("cst")));

                            //测试，先不查HBase了
                            /*String sex = "1";
                            String age = "22";
                            String cst = "2000";*/

                            //判断是否满足条件
                            if (StringUtils.equals("1", sex) && Integer.parseInt(age) < 30 && Integer.parseInt(cst) > 500) {
                                //满足用户画像条件，再继续判断
                                //* 行为属性条件: 加购物车（eventId=addCart, productId=8)>=2次, 点击广告（eventId=adClick,adId=6）>=1次
                                //从状态中获取用户行为数据
                                Iterable<LogBean> logBeanIterable = logBeanState.get();
                                //记录添加购物车操作的次数
                                int addCartCount = 0;
                                //记录点击广告行为的次数
                                int adClickCount = 0;
                                for (LogBean bean : logBeanIterable) {
                                    //判断加购物车事件
                                    if (StringUtils.equals("addCart", bean.getEventId())) {
                                        Map<String, String> propsMap = bean.getProperties();
                                        String productId = propsMap.get("productId");
                                        //判断商品Id是否为8
                                        if (StringUtils.equals("8", productId)) {
                                            addCartCount += 1;
                                        }
                                    }

                                    //判断点击广告事件
                                    if (StringUtils.equals("adClick", bean.getEventId())) {
                                        Map<String, String> propsMap = bean.getProperties();
                                        String adId = propsMap.get("adId");
                                        //判断广告Id是否为6
                                        if (StringUtils.equals("6", adId)) {
                                            adClickCount += 1;
                                        }
                                    }
                                }

                                if (addCartCount >= 2 && adClickCount >= 1) {
                                    // 满足行为属性条件，再继续判断
                                    // * 行为次序条件：依次做过 加购物车（eventId=addCart, productId=8) -> 搜索事件(eventId=search)

                                    //创建一个List，把所有行为数据都放到这个List中
                                    ArrayList<LogBean> logBeanList = new ArrayList<>();
                                    CollectionUtils.addAll(logBeanList, logBeanIterable.iterator());

                                    //遍历List
                                    StringBuilder stepBuilder = new StringBuilder();
                                    for (int i = 0; i < logBeanList.size(); i++) {
                                        LogBean bean = logBeanList.get(i);

                                        if (StringUtils.equals("addCart", bean.getEventId())) {
                                            Map<String, String> propsMap = bean.getProperties();
                                            String productId = propsMap.get("productId");
                                            //判断商品Id是否为8
                                            if (StringUtils.equals("8", productId)) {
                                                stepBuilder.append("addCart,");
                                            }
                                        }

                                        if (StringUtils.equals("search", bean.getEventId())) {
                                            stepBuilder.append("search,");
                                        }
                                    }

                                    logger.info("stepBuilder:" + stepBuilder.toString());

                                    //判断是否满足行为次序条件
                                    String pattern = ".*addCart{1,}.*search{1,}.*";
                                    boolean isMatch = Pattern.matches(pattern, stepBuilder.toString());
                                    if (isMatch) {
                                        //满足所有条件，可以输出了
                                        out.collect(new ResultBean("rule-test-001", logBean.getAccount(), logBean.getTimeStamp()));
                                    }
                                }

                            }

                        }

                    }

                    @Override
                    public void close() throws Exception {
                        //关闭hbase连接
                        if (table != null) {
                            table.close();
                        }
                        if (conn != null) {
                            conn.close();
                        }
                    }
                });

        resultDataStream.print();

        //启动程序
        env.execute("RuleEngineV1");
    }
}
